package com.bw.fun;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.ververica.cdc.debezium.DebeziumDeserializationSchema;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;

import java.util.List;

public class MyJsonSchema implements DebeziumDeserializationSchema<String> {
    @Override
    public void deserialize(SourceRecord sourceRecord, Collector<String> collector) throws Exception {
        //db tb op data
        JSONObject jo = new JSONObject();
        String[] topics = sourceRecord.topic().split("\\.");
        String db = topics[1];
        String tb = topics[2];

        Struct value = (Struct)sourceRecord.value();
        String op = value.getString("op");
        Struct after = value.getStruct("after");
        JSONObject data = new JSONObject();
        if(after!=null){
            List<Field> fields = after.schema().fields();
            for(Field f: fields){
                data.put(f.name(),after.get(f));
            }
        }

        jo.put("db",db);
        jo.put("tb",tb);
        jo.put("op",op);
        jo.put("data",data);

        collector.collect(jo.toString());
    }

    @Override
    public TypeInformation<String> getProducedType() {
        return BasicTypeInfo.STRING_TYPE_INFO;
    }
}
